package join;


import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;
import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;


import java.util.*;

/**
 * @author zhengjiasen
 * @date 2018/3/3
 */
public class Join {

    public static ArrayList<LinkedList> jobsLists = new ArrayList<LinkedList>();
    public static LinkedList<ControlledJob> cntList = new LinkedList<ControlledJob>();
    public static LinkedList<Path> pathList = new LinkedList<Path>();


    public static void main(String[] args) throws Exception {

        Join.jobsLists.add(0, cntList);
        Join.jobsLists.add(1, pathList);
        JobConf conf = new JobConf(Job.class);

        conf.set("mapreduce.output.textoutputformat.separator", ",");
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 3) {
            System.err.println("Usage: <in> [<in>...]  <out>");
            System.exit(2);
        }


        // 生成作业链表，作业个数为输出的文件个数-1（所有参数-2，因为最后一个参数为输出）。
        // 作业个数应为参数个数-2. 作业标志应为参数总数-3即i
        for (int i = 0; i < otherArgs.length - 2; i++) {
            System.out.println("当前i为:" + i);
            Job temp = Job.getInstance(conf, "join" + i);
            temp.setJarByClass(Join.class);
            //指向最后的参数。
            String out_cache = otherArgs[args.length - 1];
            if (i == 0) {
                //如果是第一个作业，则输入文件1应该为，参数1，否则是上个作业的输出缓存目录。
                System.out.println("输入文件1为：" + otherArgs[i]);
                MultipleInputs.addInputPath(temp, new Path(otherArgs[i]), TextInputFormat.class, SrcDataFileMapper.class);
            } else {
                // 上个作业的缓存目录定位时i应-1,因为保存时到下次循环，i自增了1.
                System.out.println("输入文件1为：" + out_cache + (i - 1) + "temp");
                MultipleInputs.addInputPath(temp, new Path(out_cache + (i - 1) + "temp"), TextInputFormat.class, SrcDataFileMapper.class);
            }
            //无论哪个作业，第二个输入文件应该为当前作业i +1.
            System.out.println("输入文件2为：" + otherArgs[i + 1]);
            MultipleInputs.addInputPath(temp, new Path(otherArgs[i + 1]), TextInputFormat.class, TarDataFileMapper.class);

            if (i == otherArgs.length - 3) {
                //如果是最后的作业标志，则输出文件夹为参数最后一个。否则参数+当前作业标识+"temp"
                System.out.println("最后的输出路径为：" + out_cache);
                FileOutputFormat.setOutputPath(temp, new Path(out_cache));
            } else {
                System.out.println("输出文件：" + out_cache + i + "temp");
                FileOutputFormat.setOutputPath(temp, new Path(out_cache + i + "temp"));
            }

            temp.setReducerClass(JoinReducer.class);
            //map阶段的输出的key
            temp.setMapOutputKeyClass(Text.class);
            //map阶段的输出的value
            temp.setMapOutputValueClass(Text.class);
            //reduce阶段的输出的key
            temp.setOutputKeyClass(Text.class);
            //reduce阶段的输出的value
            temp.setOutputValueClass(Text.class);
            //加入控制容器
            ControlledJob ctrljob1 = new ControlledJob(conf);
            ctrljob1.setJob(temp);

            try {
                //在作业和path 数组中取出作业链表，建立作业线性依赖关系。但第一个作业无依赖，报异常和捕捉异常。
                //原理为当前作业实例，依赖链表中最后一个作业实例。
                LinkedList test = jobsLists.get(0);
                ctrljob1.addDependingJob((ControlledJob) test.getLast());
            } catch (NoSuchElementException e) {
                System.err.println(e);
                System.err.println("第一个作业无依赖文件！");
            } catch (IndexOutOfBoundsException e) {
                System.err.println(e);
            }
            //将当前作业控制实例添加到 作业链表中。
            jobsLists.get(0).addLast(ctrljob1);
            //将当前作业缓存路径添加到缓存列表中。以便作业结束清楚缓存。
            if (i == args.length - 3) {
                jobsLists.get(1).addLast(new Path(out_cache));
            } else {
                jobsLists.get(1).addLast(new Path(out_cache + (i) + "temp"));
            }
        }


        JobControl jobCtrl = new JobControl("myctrl");

        for (Object c : jobsLists.get(0)) {
            jobCtrl.addJob((ControlledJob) c);
            System.out.println("作业队列：" + (ControlledJob) c);
        }
        for (Object c : jobsLists.get(1)) {
            System.out.println("缓存路径：" + ((Path) c).toString());
        }

        //启动线程
        Thread t = new Thread(jobCtrl);
        t.start();
        while (true) {
            //如果作业成功完成，就打印成功作业的信息
            if (jobCtrl.allFinished()) {
                System.out.println(jobCtrl.getSuccessfulJobList());
                jobCtrl.stop();
                break;
            }
        }

        FileSystem fs = FileSystem.get(conf);
        for (int p = 0; p < jobsLists.get(1).size() - 1; p++) {
            fs.delete((Path) jobsLists.get(1).get(p), true);
        }

    }
}
